package com.xian.spark.jdbc;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.*;

import java.util.HashMap;
import java.util.Properties;

/**
 * spark 2.x 读取mysql数据写到文本文件中
 */
public class ReadMysqlToFile {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("ReadMysqlToFile");
        conf.setMaster("local");

        SparkContext sc = new SparkContext(conf);

        SparkSession session = new SparkSession(sc);

        DataFrameReader read = session.read();

        //mysql配置
        HashMap<String, String> options = new HashMap<>();
        options.put("driver","com.mysql.jdbc.Driver");
        //options.put("url","jdbc:mysql://r71:3306/testdb?characterEncoding=utf8&useSSL=TRUE");
        options.put("url","jdbc:mysql://r71:3306/testdb?characterEncoding=utf8&useSSL=TRUE");
        options.put("user","root");
        options.put("password","123456");
        //options.put("dbtable","person");
        options.put("dbtable","testdb.person");

        Dataset<Row> tableDS = read.options(options).format("jdbc").load();


        tableDS.registerTempTable("t_person");
        Dataset<Row> sqlDS = session.sql("select name,birthday from t_person");
        //DataFrameWriter<Row> write = sqlDS.write();
        DataFrameWriter<Row> write = tableDS.write();

        //把Dataset写入到mysql

        Properties pro = new Properties();
        pro.setProperty("user","root");
        pro.setProperty("password","123456");

        //指定入库位追加模式SaveMode.Append   SaveMode.Overwrite会改变表结构(它会创建一模一样的表，即使mysql表不存在)
        //write.mode(SaveMode.Overwrite).jdbc("jdbc:mysql://r71:3306/db1?characterEncoding=utf8&useSSL=TRUE","t_user", pro);
        write.mode(SaveMode.Overwrite).jdbc("jdbc:mysql://r71:3306/db1?characterEncoding=utf8&useSSL=TRUE","t_user2", pro);


        session.close();
        sc.stop();


    }


}
